package correlationOfSalAndExp;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class CorrelationReducer extends Reducer<Text, Text, Text, Text> {

    private double sumXY = 0;
    private double sumX = 0;
    private double sumY = 0;
    private double sumX2 = 0;
    private double sumY2 = 0;
    private int count = 0;

    Text outValue = new Text();

    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        for (Text value : values) {
            String[] fields = value.toString().split(",");
            double X = Double.parseDouble(fields[0]);
            double Y = Double.parseDouble(fields[1]);

            // 公式里面要用到的数
            sumXY += X * Y;
            sumX += X;
            sumY += Y;
            sumX2 += X * X;
            sumY2 += Y * Y;
            count++;
        }

        // 计算皮尔逊相关系数
        double upperN = sumXY - (sumX * sumY / count);
        double underN = Math.sqrt((sumX2 - (sumX * sumX / count)) * (sumY2 - (sumY * sumY / count)));

        double correlation = upperN / underN;

        outValue = new Text("工资与经验要求相关性（皮尔逊系数）： " + correlation);

        // 输出相关性结果
        context.write(key, outValue);
    }
}
